/*
 *  Queen Mary University of London MOM System
 *  
 *  Copyright Queen Mary University of London
 *  
 *  Authors:
 *  Bob Chew (bob.chew@elec.qmul.ac.uk)
 *  Beatriz Viñal Murciano (beavimu@gmail.com)
 *  John Bigham (john.bigham@elec.qmul.ac.uk)
 *  
 *  
 *  Portions of this software uses Apache Qpid
 *    http://qpid.apache.org
 *  
 *  Apache Qpid license:
 * --------------------------------------------------------------
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 * -------------------------------------------------------------
 * 
 */
package brokerFailureDetector;

import java.util.HashMap;
import java.util.Map;
import java.util.Set;

import momControlClient.MOMControlClient;

import org.apache.qpid.client.TopicPublisherAdapter;


/**
 * Get the brokers which should be working and create threads to check if
 * each one of them is working well.
 * Send a message to the controller when the working state (=health) of a
 * broker changes.
 */
public class Detector extends MOMControlClient {

    /** Machine in which the detector is running. */
    private final String hostIP;

    /**
     * List of machines in which a broker is running and their state.
     * State is an int from WORST_STATE (worst) to BEST_STATE (best)
     */
    private HashMap<String, Integer> brokersState
            = new HashMap<String, Integer>();
    /** Worst state of a broker. */
    private static final int WORST_STATE = 0;

    /** Machine in which the controller is running. */
    private final String controllerIP;
    /** Topic used to communicate with the controller. */
    private static final String TOPIC_CONTROLLER = "brokerFailure.controller";
    /** Publisher which will send messages to the controller. */
    private TopicPublisherAdapter publisherToController;

    /**
     * Constructor.
     *
     * @param hostIP  IP address of the machine the detector is running on
     * @param controllerIP  IP address of the controller
     * @param brokerAddresses  list of machines in which a broker is running
     *                         and the port in which they are listening
     */
    public Detector(final String hostIP, final String controllerIP,
                    final HashMap<String, Integer> brokerAddresses) {

        super(); //assign ID

        this.hostIP = hostIP;
        this.controllerIP = controllerIP;

        //start a connection to send messages to the controller
        try {
            //create and start a connection and a session in it
            openConnAndSess(this.controllerIP);
            //create publisher
            publisherToController = createPublisher(TOPIC_CONTROLLER);
        } catch (Exception e) {
            e.printStackTrace();
        }

        Set<Map.Entry<String, Integer>> brkAddrSet = brokerAddresses.entrySet();

        //for each broker
        for (Map.Entry<String, Integer> entry : brkAddrSet) {
             //put its initial state (off) in brokersState
             brokersState.put(entry.getKey(), WORST_STATE);
             //create a thread
             new DetectorThread(this, entry.getKey(), entry.getValue());
        }

         System.out.println("Detector in " + this.hostIP + " ready");

    }

    /**
     * Receives the most recent state of a broker. If it is not the same as the
     * previous state of that broker, informs the controller about the change.
     *
     * @param broker  whose state is being updated
     * @param state  new state of the broker
     */
     protected void updateState(final String broker, final int state) {
        //get the previous state
        int oldState = brokersState.get(broker);
        //save the new state
        brokersState.put(broker, state);
        //decide when to report a change to the controller
        if (oldState != state) {
            System.out.println("The detector in " + hostIP + " detected that "
                               + "the broker in " + broker + " changed its "
                               + "state from " + oldState + " to " + state);
            //send message to broker
            //create the text of the message to send
            String text = hostIP + ":" + broker + ":" + state;
            try {
                //send the message to the controller
                sendMessage(publisherToController, text);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
//        System.out.println("State in " + HOST_BROKER + ": " + brokersState);
    }

}
